Description

When running multi-DUT tests, the output from different DUTs can be interleaved/garbled when two devices print at the same time. For example:

2026-04-30 23:47:40 [dut-0] [SERVER] DescPerm WriteAuthorized-no-write re2026-04-30 23:47:40 [dut-1] [CLIENT] Reconnected for introspection
jected: 1

This happens because each DUT has its own _listen function running in a separate multiprocessing.Process (spawned via _ctx = multiprocessing.get_context('spawn')), and they all write to the same _stdout (sys.__stdout__) with no synchronization. Two processes can interleave their _stdout.write() + _stdout.flush() calls. This can cause expect to fail.

This pull request introduces a robust solution to prevent garbled or interleaved output when multiple DUT (Device Under Test) listener processes write to stdout concurrently. It does so by adding a session-scoped multiprocessing lock that is shared among all listener processes, ensuring serialized writes to stdout. The changes include the implementation of the lock, integration into the DUT factory and plugin, and comprehensive tests to verify data integrity in concurrent scenarios.

Stdout Locking and Listener Process Improvements:

• Introduced a session-scoped multiprocessing lock (_stdout_lock) to serialize stdout writes across all DUT listener processes, preventing garbled output when multiple DUTs print simultaneously. The lock is set up as a pytest fixture with autouse=True, ensuring it is always available and registered before any DUT is instantiated. (pytest-embedded/pytest_embedded/plugin.py, pytest-embedded/pytest_embedded/dut_factory.py) [1] [2] [3]
• Modified the _listen function and related process creation logic to accept and use the shared _stdout_lock, wrapping stdout writes with the lock when present. This ensures atomic output and prevents interleaving from concurrent processes. (pytest-embedded/pytest_embedded/dut_factory.py) [1] [2] [3]
• Added set_stdout_lock utility to update the module-level lock reference, facilitating global access for all listener processes. (pytest-embedded/pytest_embedded/dut_factory.py, pytest-embedded/pytest_embedded/plugin.py) [1] [2]

Testing and Validation:

• Added a comprehensive test suite to ensure that the stdout lock mechanism prevents data loss and output corruption, even under concurrent access by multiple DUTs. Tests cover scenarios with and without the lock, concurrent listeners, and multi-DUT pytest runs. (pytest-embedded/tests/test_base.py)

Internal Refactoring:

• Updated imports and internal function signatures to support the new locking mechanism, ensuring seamless integration with existing DUT and plugin infrastructure. (pytest-embedded/pytest_embedded/plugin.py) [1] [2]

These changes collectively improve the reliability and clarity of test output in multi-DUT environments, especially when tests are run in parallel.

Testing

Tested locally. It solves the issue and work reliably with both single and multi dut tests.